package data.FlinkJoin

import org.apache.flink.streaming.api.functions.source.{RichParallelSourceFunction, SourceFunction}

import scala.collection.mutable.ListBuffer

/**
 * 名字,性别,时间戳
 **/
class ASource extends RichParallelSourceFunction[(String,String,Long)]{
  var running = true

  override def run(ctx: SourceFunction.SourceContext[(String, String, Long)]): Unit = {
    val buffer = ListBuffer[(String, String, Long)]()
    buffer.+=(("PK","M",50000L)) // [50,60)
    buffer.+=(("PK","M",54000L))// [50,60)
    buffer.+=(("PK","M",79900L))  // [70,80)
    buffer.+=(("PK","M",115000L))  // [110,120)
    buffer.+=(("RZ","M",100000L))  // [100,110)
    buffer.+=(("RZ","M",109000L)) // [100,110)

    var count = 0
    while(running && count < buffer.length) {
      ctx.collect((buffer(count)._1, buffer(count)._2, buffer(count)._3))
      count += 1

      Thread.sleep(1000)
    }
  }

  override def cancel(): Unit = {
    running = false
  }
}
